package david.java.flink.window;

import david.java.flink.datasources.Tuple3Source;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.EnvironmentSettings;
import scala.Tuple3;

/**
 * @Description:
 * @Author: David
 * @Date: Create in 7:31 下午 2022/1/12
 */
public class SlidingWindowTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();

        DataStreamSource<Tuple3<String, Long, Long>> data = env.addSource(new Tuple3Source());

        // 用来检验sliding window 是如何触发计算的
        data.keyBy(t -> t._1())
                .window(SlidingProcessingTimeWindows.of(Time.seconds(20), Time.seconds(10)))
                .reduce((ReduceFunction<Tuple3<String, Long, Long>>) (value1, value2) -> value1._2() > value2._2() ? value1 : value2)
                ;
                // .map()
                // .print();



        env.execute("sliding window test");

    }
}
